package avicit.bdp.dds.server.worker.processor;

import avicit.bdp.common.dto.PythonResourceDTO;
import avicit.bdp.common.service.service.FileOperateCommonService;
import avicit.bdp.common.utils.ConfigUtils;
import avicit.bdp.common.utils.SpringApplicationContext;
import avicit.bdp.common.utils.file.StorageConfThreadLocal;
import avicit.bdp.dds.apc.dto.PythonResourceLogDTO;
import avicit.bdp.dds.apc.service.PythonResService;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.common.thread.ThreadUtils;
import avicit.bdp.dds.common.utils.OSUtils;
import avicit.bdp.dds.common.utils.Preconditions;
import avicit.bdp.dds.remote.command.Command;
import avicit.bdp.dds.remote.command.CommandType;
import avicit.bdp.dds.remote.command.ShellExecuteRequestCommand;
import avicit.bdp.dds.remote.processor.NettyRequestProcessor;
import avicit.bdp.dds.remote.utils.FastJsonSerializer;
import avicit.bdp.dds.server.worker.cache.TaskExecutionContextCacheManager;
import avicit.bdp.dds.server.worker.cache.impl.TaskExecutionContextCacheManagerImpl;
import avicit.bdp.dds.service.process.ProcessService;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.function.Consumer;

/**
 * execute shell processor
 */
public class ShellExecuteProcessor implements NettyRequestProcessor {

    private final Logger logger = LoggerFactory.getLogger(ShellExecuteProcessor.class);
    private static final String MIRRORS = "https://pypi.tuna.tsinghua.edu.cn/simple ";

    private final FileOperateCommonService fileOperateCommonService;
    /**
     *  process database access
     */
    private final ProcessService processService;
    /**
     * task callback service
     */
    private final TaskCallbackService taskCallbackService;
    /**
     * taskExecutionContextCacheManager
     */
    private TaskExecutionContextCacheManager taskExecutionContextCacheManager;
    private final PythonResService pythonResService;
    /**
     * log list
     */
    protected final List<String> logBuffer;
    /**
     * log handler
     */
    protected Consumer<LogVo> logHandler;

    class LogVo{
        private String logId;
        private String resId;
        private List<String> logs;

        public String getLogId() {
            return logId;
        }

        public void setLogId(String logId) {
            this.logId = logId;
        }

        public String getResId() {
            return resId;
        }

        public void setResId(String resId) {
            this.resId = resId;
        }

        public List<String> getLogs() {
            return logs;
        }

        public void setLogs(List<String> logs) {
            this.logs = logs;
        }
    }

    public ShellExecuteProcessor() {
        this.fileOperateCommonService = SpringApplicationContext.getBean(FileOperateCommonService.class);
        this.processService = SpringApplicationContext.getBean(ProcessService.class);
        this.taskCallbackService = SpringApplicationContext.getBean(TaskCallbackService.class);
        this.taskExecutionContextCacheManager = SpringApplicationContext.getBean(TaskExecutionContextCacheManagerImpl.class);
        this.pythonResService = SpringApplicationContext.getBean(PythonResService.class);
        this.logBuffer = Collections.synchronizedList(new ArrayList<>());
        this.logHandler = this::logHandle;
    }

    public void logHandle(LogVo logVo) {
        // note that the "new line" is added here to facilitate log parsing
        String logs = String.join("\n\t", logVo.getLogs());
        logger.info(" -> {}", logs);

        PythonResourceLogDTO pythonResourceLog = pythonResService.getResourceLogById(logVo.getLogId());
        if(pythonResourceLog != null){
            pythonResService.updateResourceLogById(pythonResourceLog, logs);
        }else{
            pythonResService.inserResourceLog(logVo.getLogId(), logs);
        }

        // 安装成功
        if(logs.contains("Successfully installed") || logs.contains("is already installed")){
            processService.updatePythonResourceById(logVo.getResId(), 1);
        }
    }

    /**
     * @param channel channel
     * @param command command
     */
    @Override
    public void process(Channel channel, Command command) {
        Preconditions.checkArgument(CommandType.SHELL_EXECUTE_REQUEST == command.getType(), String.format("invalid command type : %s", command.getType()));

        ShellExecuteRequestCommand requestCommand = FastJsonSerializer.deserialize(command.getBody(), ShellExecuteRequestCommand.class);
        logger.info("received execute shell request command : {}", requestCommand);

        Process process = null;
        // 在线安装
        if(requestCommand.getOnline() == 1){
            process = onlineInstall(requestCommand);
        }else{
            // 离线安装
            process = offlineInstall(requestCommand);
        }

        parseProcessOutput(process, requestCommand.getLogId(), requestCommand.getResourceId());
    }

    private Process onlineInstall(ShellExecuteRequestCommand requestCommand) {
        // 执行脚本安装命令
        String pyScript = requestCommand.getPyScript();
        Process process = null;
        try {
            if (OSUtils.isWindows()) {
                process = OSUtils.exeCommand(pyScript);
            }else{
                process = OSUtils.exeShellCommand(pyScript);
            }
        }catch (IOException e) {
            logger.error("执行在线安装python包失败", e);
        }
        return process;
    }

    private Process offlineInstall(ShellExecuteRequestCommand requestCommand) {
        // 下载资源包,并执行安装脚本
        String resourceId = requestCommand.getResourceId();
        PythonResourceDTO resource = processService.getPythonResourceById(resourceId);
        if (resource == null) {
            logger.error("python resource id: {} not exist", resourceId);
            throw new RuntimeException(String.format("python resource id: %s not exist", resourceId));
        }
        String resourceName = resource.getFullName().replaceFirst(Constants.SLASH, "");
        logger.info("resourceName : {}", resourceName);
        Process process = null;

        try {
            // 下载安装包
            String dstFile = "/tmp/" + resourceName;
            if(dstFile.indexOf("/") == 0){
                dstFile = dstFile.substring(1);
            }
            fileOperateCommonService.copyServerToLocal(resource.getFullName(), dstFile, false, true);

            File dstPath = new File(dstFile);
            if(!dstPath.exists()){
                logger.error("安装文件下载异常 : {}", dstFile);
                throw new RuntimeException("安装文件下载异常");
            }

            // 执行脚本安装命令
            String pyScript = ConfigUtils.getInstance().getString("pip.cmd") + " install -i " + MIRRORS + dstFile;
            if (OSUtils.isWindows()) {
                process = OSUtils.exeCommand(pyScript);
            }else{
                process = OSUtils.exeShellCommand(pyScript);
            }
        } catch (IOException e) {
            logger.error("执行离线安装python包失败", e);
        } finally {
            StorageConfThreadLocal.removeThreadLocal();
        }
        return process;
    }

    private void parseProcessOutput(Process process, String logId, String resId) {
        logger.info("parseProcessOutput logId={}, resId={}", logId, resId);
        String threadLoggerInfoName = "Python Install Log";
        ExecutorService parseProcessOutputExecutorService = ThreadUtils.newDaemonSingleThreadExecutor(threadLoggerInfoName);
        parseProcessOutputExecutorService.submit(() -> {
            BufferedReader inReader = null;
            try {
                if (OSUtils.isWindows()) {
                    inReader = new BufferedReader(new InputStreamReader(process.getInputStream(), "GBK"));
                } else {
                    inReader = new BufferedReader(new InputStreamReader(process.getInputStream(), "UTF-8"));
                }

                long lastFlushTime = System.currentTimeMillis();
                String line;
                while ((line = inReader.readLine()) != null) {
                    logBuffer.add(line);
                    lastFlushTime = flush(lastFlushTime, logId, resId);
                }
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            } finally {
                clear(logId, resId);
                close(inReader);
            }
        });
        parseProcessOutputExecutorService.shutdown();
    }

    /**
     * when log buffer siz or flush time reach condition , then flush
     *
     * @param lastFlushTime last flush time
     * @return last flush time
     */
    private long flush(long lastFlushTime, String logId, String resId) {
        long now = System.currentTimeMillis();
        // when log buffer siz or flush time reach condition , then flush
        if (logBuffer.size() >= Constants.DEFAULT_LOG_ROWS_NUM || now - lastFlushTime > Constants.DEFAULT_LOG_FLUSH_INTERVAL) {
            lastFlushTime = now;
            // log handle

            LogVo logVo = new LogVo();
            logVo.setLogs(logBuffer);
            logVo.setLogId(logId);
            logVo.setResId(resId);
            logHandler.accept(logVo);

            logBuffer.clear();
        }
        return lastFlushTime;
    }

    /**
     * clear
     */
    private void clear(String logId, String resId) {
        if (!logBuffer.isEmpty()) {
            // log handle

            LogVo logVo = new LogVo();
            logVo.setLogs(logBuffer);
            logVo.setLogId(logId);
            logVo.setResId(resId);
            logHandler.accept(logVo);

            logBuffer.clear();
        }
    }

    /**
     * close buffer reader
     *api/bdp/dcs/file
     * @param inReader in reader
     */
    private void close(BufferedReader inReader) {
        if (inReader != null) {
            try {
                inReader.close();
            } catch (IOException e) {
                logger.error(e.getMessage(), e);
            }
        }
    }

}
